package com.zhongyu.scheduler.consumer.service.impl;

import com.zhongyu.comm.pojo.Task;
import com.zhongyu.comm.util.JacksonUtil;
import com.zhongyu.comm.vo.CreateOrderMessage;
import com.zhongyu.scheduler.consumer.client.TaskClient;
import com.zhongyu.scheduler.consumer.service.PublishMessageService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.connection.RabbitUtils;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.List;

@Service("publishMessageService")
@Slf4j
public class PublisMessageServiceImpl implements PublishMessageService ,RabbitTemplate.ConfirmCallback{

    @Resource
    private RabbitTemplate rabbitTemplate;



    @Resource
    private TaskClient taskClient;

    @Override
    public void confirm(CorrelationData correlationData, boolean b, String s) {
      String taskId= correlationData.getId();
      if (b==true){
        taskClient.updateStatus(taskId,1,LocalDateTime.now());
      }else {
         taskClient.updateErrormsg(taskId,s,LocalDateTime.now());
      }
    }

    private void publishTask(Task task, String ex, String routingKey) {

        //设置消息唯一id
        CorrelationData correlationData = new CorrelationData(task.getId());
      // rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
       rabbitTemplate.setConfirmCallback(this);
        //将消息发送到相应的交换器
        rabbitTemplate.convertAndSend(ex,routingKey,task,correlationData);

    }

    //项目启动后三秒钟，每隔10秒钟执行一次任务
    @Override
   @Scheduled(initialDelay = 3000,fixedDelay = 10000)
    public void tranferTesk() {
       //取出前100状态为0或未发送或未成功的信息
        List<Task> list=taskClient.findByStatusAndOvertimeBefore(0, LocalDateTime.now(),100);
        if (!list.isEmpty()){
            list.forEach(task -> {
                //采用乐观锁，保证在多节点集群状态下，该对象只处理一次
                //乐观锁机制，每次取到消息都更新消息版本号，保证一条消息不会被多个节点同时发送
               int result= taskClient.updateTaskVersion(task.getId(),task.getVersion());
                //更新成功即说明该条消息，本节点能够发送
                if (result>0){
                    //一条消息尝试三次以上即表示失败，不再发送，通过人工处理
                    if (task.getTryCount()>=3){
                        taskClient.updateStatus(task.getId(),2,LocalDateTime.now());
                    }else {
                        taskClient.updateTryCountByTaskId(task.getId(),LocalDateTime.now());
                        try {
                            log.debug("开始发送创建订单信息!");
                        /* String requestBody= task.getRequestBody();
                          CreateOrderMessage createOrderMessage= (CreateOrderMessage) JacksonUtil.json2Obj(requestBody);
                          createOrderMessage*/
                            this.publishTask(task,task.getMqExchange(),task.getMqRoutingkey());
                        }catch (Exception ex){
                                  log.error("创建订单信息发送失败!",ex);
                        }

                    }
               }

            });
        }
    }


}
